#include "msg_queue_peer.h"
#include "shmem.h"
#include "cJSON.h"
#include <pthread.h>
#include "MQTTClient_fun.h"

MQTTClient client;//客户端
struct shm_param shm_para;//用于共享内存的结构体

char  *addr = NULL;//共享内存地址,数据时实变化

char  json_str [1024] = {0};//承接来自共享内存的数据

SensorData St_data;//承接解析完的数据

pthread_mutex_t json_lock;

//pthread_mutex_t mqtt_lock;

int mode;//1循环    2刷新   //对比

uint8_t refresh_flag = 0;//刷新标志位，为1时上报数据

uint8_t msg_send_flag = 0;//控制命令消息队列发送标志位

SensorData  last_data     =   {0};

MsgQueue msg_payloadstr = {0};

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)//回调函数，处理QT端发过来的数据
{
    char  payloadptr[1024];

    //printf("Message arrived\n");
    //printf("     topic: %s\n", topicName);
    //printf("   message: ");

    strcpy(payloadptr, message->payload);
    payloadptr[message->payloadlen] = '\0';
     
    if (message->payloadlen > 0) {
        //printf("%s\n", payloadptr);
    } else {
        printf("(null)\n");
    }
    char   cmd[512] = {0};
    Empty_Control(payloadptr,cmd);

   // printf("2222222222222222222\n");

    if(strcmp(cmd,"cmd")  == 0)       //如果为空代表是控制外设命令，传输给stm32
    {
        
        memset(&msg_payloadstr,0,sizeof(msg_payloadstr));//初始化

        msg_payloadstr.msg_type = 1;//赋值
        strcpy(msg_payloadstr.data,payloadptr);
        msg_payloadstr.data_len = strlen(payloadptr);

        msg_send_flag = 1;
       // printf("payloadlen = %d\n",message->payloadlen);
    }
    else
    {
        //printf("change mode to : %s\n",cmd);
        if(strcmp(cmd,"cycle")  == 0)   //不为空代表为模式切换命令，判断改变模式
        {
            mode    =   CYCLE_MODE;
        }
        else    if(strcmp(cmd,"refresh") ==  0)
        {
            mode    =   REfresh_MODE;
            refresh_flag    = 1;
			//printf("******************\n");
        }
        else    if(strcmp(cmd,"threshold") ==  0)
        {
            mode    =   Contrast_MODE;
        }
        else
        {
            printf("Invalid mode\n");
        }
      //  free(cmd);
       // cmd = NULL;
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

// 将结构体转换为cJSON字符串
char* serialize_sensor_data(const SensorData* data) 
{
    if (data == NULL) {
        return NULL;  // 若输入数据为空，则返回NULL
    }
 
    // 创建一个cJSON对象
    cJSON *root = cJSON_CreateObject();
    if (root == NULL) {
        return NULL;  // 若对象创建失败，则返回NULL
    }
 
    // 向cJSON对象中添加数据
    cJSON_AddNumberToObject(root, "temperature", data->temperature);
    cJSON_AddNumberToObject(root, "humidity", data->humidity);
    cJSON_AddNumberToObject(root, "light", data->light);
    cJSON_AddNumberToObject(root, "fan", data->fan);
    cJSON_AddNumberToObject(root, "motor", data->motor);
    cJSON_AddNumberToObject(root, "buzzer", data->buzzer);
    cJSON_AddNumberToObject(root, "LED1", data->LED1);
    cJSON_AddNumberToObject(root, "LED2", data->LED2);
    cJSON_AddNumberToObject(root, "LED3", data->LED3);
    cJSON_AddNumberToObject(root, "LED4", data->LED4);
   // cJSON_AddNumberToObject(root, "updated", data->updated);
 
    // 将cJSON对象转换为字符串
    char *json_string = cJSON_Print(root);
    if (json_string == NULL) {
        cJSON_Delete(root);  // 若字符串转换失败，则删除cJSON对象并返回NULL
        return NULL;
    }
 
    // 清理cJSON对象（注意：json_string包含了数据的副本，因此可以安全删除原始对象）
    cJSON_Delete(root);
 
    return json_string;  // 返回JSON字符串
}

// 将cJSON字符串转换为结构体
int deserialize_sensor_data(const char* json_string, SensorData* data) {
    // Parse the JSON string
    cJSON *root = cJSON_Parse(json_string);
    if (root == NULL) {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL) {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        return -1; // Parsing failed
    }
    
    cJSON *temperature_item = cJSON_GetObjectItem(root, "temperature");
    if (temperature_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(temperature_item->type == cJSON_Number) {
        data->temperature = temperature_item->valuedouble;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    cJSON *humidity_item = cJSON_GetObjectItem(root, "humidity");
    if (humidity_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(humidity_item->type == cJSON_Number) {
        data->humidity = humidity_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    cJSON *light_item = cJSON_GetObjectItem(root, "light");
    if (light_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(light_item->type == cJSON_Number) {
        data->light = light_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    cJSON *fan_item = cJSON_GetObjectItem(root, "fan");
    if (fan_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(fan_item->type == cJSON_Number) {
        data->fan = fan_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    cJSON *motor_item = cJSON_GetObjectItem(root, "motor");
    if (motor_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(motor_item->type == cJSON_Number) {
        data->motor = motor_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    cJSON *buzzer_item = cJSON_GetObjectItem(root, "buzzer");
    if (buzzer_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(buzzer_item->type == cJSON_Number) {
        data->buzzer = buzzer_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    cJSON *LED1_item = cJSON_GetObjectItem(root, "LED1");
    if (LED1_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(LED1_item->type == cJSON_Number) {
        data->LED1 = LED1_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    cJSON *LED2_item = cJSON_GetObjectItem(root, "LED2");
    if (LED2_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(LED2_item->type == cJSON_Number) {
        data->LED2 = LED2_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    cJSON *LED3_item = cJSON_GetObjectItem(root, "LED3");
    if (LED3_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(LED3_item->type == cJSON_Number) {
        data->LED3 = LED3_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    cJSON *LED4_item = cJSON_GetObjectItem(root, "LED4");
    if (LED4_item == NULL) {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }
    if(LED4_item->type == cJSON_Number) {
        data->LED4 = LED4_item->valueint;
    } else {
        cJSON_Delete(root);
        printf("Failed to parse JSON data.\n");
        return -1; // Parsing failed
    }

    /*
    // Get and set the sensor data fields
    data->temperature = cJSON_GetObjectItem(root, "temperature")->valueint;
    data->humidity = cJSON_GetObjectItem(root, "humidity")->valueint;
    data->light = cJSON_GetObjectItem(root, "light")->valueint;
    data->fan = cJSON_GetObjectItem(root, "fan")->valueint;
    data->motor = cJSON_GetObjectItem(root, "motor")->valueint;
    data->buzzer = cJSON_GetObjectItem(root, "buzzer")->valueint;
    data->LED1 = cJSON_GetObjectItem(root, "LED1")->valueint;
    data->LED2 = cJSON_GetObjectItem(root, "LED2")->valueint;
    data->LED3 = cJSON_GetObjectItem(root, "LED3")->valueint;
    data->LED4 = cJSON_GetObjectItem(root, "LED4")->valueint;
    //data->updated = cJSON_GetObjectItem(root, "updated")->valueint;
    */
 
    // Clean up
    cJSON_Delete(root);
 
    return 0; // Success
}

void  Empty_Control(const char *mqtt_json,char *temp)
{
  // 解析 JSON 字符串
    
    cJSON *root = cJSON_Parse(mqtt_json);
    if (root == NULL) {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL) {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        return ; // 解析失败，返回 
    }
    // 在 JSON 对象中查找 "command" 字段
    cJSON *command_item = cJSON_GetObjectItem(root, "command");

    if(command_item == NULL)
    {
		cJSON_Delete(root);
		stpcpy(temp,"cmd");
        return ;
    }
    else  if (command_item->type == cJSON_String && (command_item->valuestring != NULL)) {
        // 如果找到了 "command" 字段，并且是字符串类型，则返回其值
		strcpy(temp,command_item->valuestring);
		//printf("%s\n",temp);
		cJSON_Delete(root);
		
        return ;
    } 

    // 释放 JSON 对象（注意：在实际应用中，应确保在所有退出路径上都释放内存）
    cJSON_Delete(root);
    return ;
}

bool    Compare_Sensor_data(SensorData * last,SensorData  *current )//比较结构体数据
{
    
    return  abs(current->temperature-last->temperature)>=TEMP_CONTRAST ||
            abs(current->humidity-last->humidity)>=HUM_CONTRAST ||
            abs(current->light-last->light)>=LIGHT_CONTRAST  ||
            current->fan != last->fan ||
            current->motor != last->motor ||
            current->buzzer != last->buzzer ||
            current->LED1 != last->LED1 ||
            current->LED2 != last->LED2 ||
            current->LED3 != last->LED3 ||
            current->LED4 != last->LED4;
}

// MQTT重连函数
int reconnect_mqtt() {
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    
    int rc = MQTTClient_connect(client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect to MQTT broker, rc=%d\n", rc);
        return 0;
    }
    
    // 重新订阅主题
    MQTTClient_subscribe(client, "sub_example", QOS);
    return 1;
}

void *thread_Process_Stm32toB(void *arg)//共享内存读取线程
{
    //printf("thread_Process_Stm32toB_start\n");
    while (1)
    {
      //  printf("read from shem_addr\n");
         pthread_mutex_lock(&json_lock);//上锁

	memset(json_str,0,sizeof(json_str));
        strcpy(json_str,addr);
       
        //strcpy(json_str , ("{\"temperature\":25,\"humidity\":50,\"light\":100,\"fan\":1,\"motor\":0,\"buzzer\":0,\"LED1\":1,\"LED2\":0,\"LED3\":1,\"LED4\":0}"));
        

        if(deserialize_sensor_data(json_str,&St_data) != 0)
        {
            printf("Failed to deserialize JSON data.\n");
        }
        
        pthread_mutex_unlock(&json_lock);//解锁

       // printf("read from shem_addr over\n");
       
        usleep(100000); //100mS
    }
}

void *thread_Process_BtoQT(void *arg)//mqtt发布线程
{
    while(1)
    {
        MQTTClient_message pubmsg = MQTTClient_message_initializer;
    
        pthread_mutex_lock(&json_lock);//上锁

        pubmsg.payload = json_str;
        pubmsg.payloadlen = (int)strlen(json_str);

        pthread_mutex_unlock(&json_lock);//解锁

        pubmsg.qos = QOS;
        pubmsg.retained = 0;
    
        MQTTClient_deliveryToken token;
        int rc = 0;


        if(mode == CYCLE_MODE)
        {
            rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
            if (rc != MQTTCLIENT_SUCCESS) {
                printf("Failed to publish message, return code %d\n", rc);
            }
           // printf("publication of %s\n""on topic %s for client with ClientID: %s\n",(char *)pubmsg.payload, TOPIC, CLIENTID_Pub);
          //  printf("Message with delivery token %d delivered\n", token);
            sleep(1);
        }
        else if(mode == REfresh_MODE )
        {
            if(refresh_flag)
            {
                rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
                if (rc != MQTTCLIENT_SUCCESS)
                {
                    printf("Failed to publish message, return code %d\n", rc);
                }
                refresh_flag = 0;//刷新标志位清零
            }

           // printf("publication of %s\n""on topic %s for client with ClientID: %s\n",(char *)pubmsg.payload, TOPIC, CLIENTID_Pub);
          //  printf("Message with delivery token %d delivered\n", token);

            
        }
        else if(mode == Contrast_MODE)
        {
            if(Compare_Sensor_data(&last_data,&St_data))
            {
                last_data = St_data;
                rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
                if (rc!= MQTTCLIENT_SUCCESS)
                {
                    printf("Failed to publish message, return code %d\n", rc);
                    
                }
              //  printf("publication of %s\n""on topic %s for client with ClientID: %s\n",(char *)pubmsg.payload, TOPIC, CLIENTID_Pub);
             //   printf("Message with delivery token %d delivered\n", token);
            }
        }
        else 
        {
            printf("Invalid mode Auto change mode to CYCLE \n");
            mode = CYCLE_MODE;
        }
	  if(rc != MQTTCLIENT_SUCCESS)
        {
            printf("Failed to publish message, return code %d\n", rc);
            int retry = 3;
            while(retry-- > 0) {
                sleep(1);  // 等待1秒
                if (reconnect_mqtt()) {
                    printf("Reconnected to MQTT broker\n");
                    break;
                }
            }
        }
    }

    return NULL;
}

void *pthread_MsgQueue_Ctlstm32(void * arg)//消息队列控制命令线程
{
    //printf("pthread_MsgQueue_Ctlstm32\n");
    while (1)
    {
        if(msg_send_flag)
        {
            //printf("control stm32:%s\n",msg_payloadstr.data);
            if(msg_queue_send("topic", &msg_payloadstr, sizeof(msg_payloadstr), 0) < 0)//发送消息
            {
                printf("msg_queue_send error\n");
            }
            msg_send_flag = 0;
        }
    }
}

int main(int argc, char *argv[])
{
    int ret = shm_init(&shm_para,"shm_BtoA",SHM_SIZE);//共享内存初始化
    if(ret < 0)
    {
        return -1;
    }

    addr = (char *)shm_getaddr(&shm_para);//得到共享内存
    if(addr == NULL)
    {
        printf("shmem is NULL !\n");
        return -1;
    }
    sleep(1);
    

    mode = CYCLE_MODE;//默认循环上报模式

    pthread_mutex_init(&json_lock,NULL);//互斥锁初始化
    //pthread_mutex_init(&mqtt_lock,NULL);


    MQTTClient_connectOptions conn_opts  =  MQTTClient_connectOptions_initializer;//创建并初始化客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);// 设置客户端回调，包括连接丢失、消息到达和消息送达确认的回调

    if(MQTTClient_connect(client, &conn_opts) != MQTTCLIENT_SUCCESS)//判断连接是否成功
    {
        printf("Failed to connect, return code %d\n", MQTTClient_connect(client, &conn_opts));
         int retry = 3;
        while(retry-- > 0) {
            sleep(1);  // 等待1秒
            if (reconnect_mqtt()) {
                printf("Reconnected to MQTT broker\n");
                break;
            }
        }
        if (retry <= 0) {
            return -1;
        }
    }

    MQTTClient_subscribe(client, "sub_example", QOS);//订阅主题

    pthread_t  pth_AtoB,  pth_BtoC,pth_BtoA;

    if(0 != pthread_create(&pth_AtoB, NULL, thread_Process_Stm32toB, NULL))//创建各个线程
    {
        printf("pthread_AtoB create failed\n");
    }
  
    if(0!= pthread_create(&pth_BtoC, NULL, thread_Process_BtoQT, NULL))
    {
        printf("pthread_BtoC create failed\n");
    }

    if (0 != pthread_create(&pth_BtoA,NULL,pthread_MsgQueue_Ctlstm32,NULL))
    {
        printf("pthread_BtoA create failed\n");
    }
    
    
    //printf("pthread creat over\n");
    


    if(pthread_join(pth_AtoB, NULL) != 0)//回收各线程
    {
        perror("Failed to join thread_AtoB\n");
        exit(EXIT_FAILURE);
    }

    if(pthread_join(pth_BtoC, NULL)!= 0)
    {
        perror("Failed to join thread_BtoC\n");
        exit(EXIT_FAILURE);
    }
    if(pthread_join(pth_BtoA, NULL)!= 0)
    {
        perror("Failed to join thread_BtoA\n");
        exit(EXIT_FAILURE);
    }


    shm_del(&shm_para);//解除共享内存

    MQTTClient_disconnect(client, 10000);//断开客户端连接

    MQTTClient_destroy(&client);//销毁客户端

    pthread_mutex_destroy(&json_lock);

    return 1;
}
